package com.gagakuai.app.flink.windows;

import com.gagakuai.domain.facts.Fact;
import com.gagakuai.domain.index.calculate.IndexCalculateConfig;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * @description: 窗口分配器
 * @author: houhong
 * @create: 2024-09-07 19:17
 **/
public class MetricWindowAssigner extends WindowAssigner<Fact<?>, TimeWindow> {

    /*
     * 大小
     */
    private long size;

    /*
     * 步长
     */
    private long step;

    /*
     * 偏移量
     */
    private long offset;

    /**
     * @description: 将带有时间戳的行为事件 分配给1个或多个窗口,返回的是1个单例窗口集合
     * @param fact: 配置
     * @param timestamp: 行为事件的时间戳
     * @param windowAssignerContext:  窗口上下文
     * @return java.util.Collection<org.apache.flink.streaming.api.windowing.windows.TimeWindow>
     */
    @Override
    public Collection<TimeWindow> assignWindows(Fact<?> fact, long timestamp, WindowAssignerContext windowAssignerContext) {

        IndexCalculateConfig config = (IndexCalculateConfig)fact.getValue();

        String windowSize = config.getWindow_size();
        size = Long.parseLong(windowSize) * 1000L;

        String windowStep = config.getWindow_step();
        step = Long.parseLong(windowStep) * 1000L;

        String _winEnd = config.getFlink_filter();

        if(_winEnd.equals("0")) {
            /* 没有设置窗口固定时间范围 */
            return windowNoDaysRange(timestamp);
        }

        //获取当前行为事件所属窗口的开始时间
        long lastStart = TimeWindow.getWindowStartWithOffset(
                timestamp,
                this.offset,
                this.step);

        long lastEnd = lastStart + step;

        long winEnd = Long.parseLong(_winEnd);
        return windowWithDaysRange(timestamp, lastStart, lastEnd, winEnd);
    }

    private Collection<TimeWindow> windowWithDaysRange(
            long timestamp,
            long lastStart,
            long lastEnd,
            long winEnd) {

        if (timestamp == Long.MIN_VALUE) {

            throw new RuntimeException("Record has Long.MIN_VALUE " +
                    "timestamp (= no timestamp marker). " +
                    "Is the time characteristic set to 'ProcessingTime', " +
                    "or did you forget to call 'DataStream.assignTimestampsAndWatermarks(...)'?");
        }
        //获取每条行为事件被分到的窗口数量
        int winCounts = (int) ((winEnd - lastEnd) / this.step);
        List<TimeWindow> windows = new ArrayList<>(winCounts);

        while (lastEnd < winEnd) {
            windows.add(new TimeWindow(lastStart, lastEnd));
            lastEnd += step;
        }
        return windows;
    }


    private Collection<TimeWindow> windowNoDaysRange(long timestamp) {

        if (timestamp == Long.MIN_VALUE) {
            throw new RuntimeException("Record has Long.MIN_VALUE timestamp " +
                    "(= no timestamp marker). Is the time characteristic set to " +
                    "'ProcessingTime', or did you forget to call 'DataStream.assignTimestampsAndWatermarks(...)'?");
        }

        List<TimeWindow> windows = new ArrayList((int)(this.size / this.step));
        long lastStart = TimeWindow.getWindowStartWithOffset(timestamp, this.offset, this.step);

        for(long start = lastStart; start > timestamp - this.size; start -= this.step) {
            windows.add(new TimeWindow(start, start + this.size));
        }

        return windows;
    }


    @Override
    public Trigger<Fact<?>, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment streamExecutionEnvironment) {
        return null;
    }

    @Override
    public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
        return new TimeWindow.Serializer();
    }

    @Override
    public boolean isEventTime() {
        return true;
    }
}
